#pragma once
#include "../mqcommon/logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <mutex>
#include <memory>

namespace mq_zxws
{
    using ConsumerCallBack = std::function<void(const std::string, const BasicProperties*, const std::string)>;
    struct Consumer{
        using ptr = std::shared_ptr<Consumer>;
        std::string tag;   // 消费者标识
        std::string qname; // 订阅的消息队列
        bool auto_ack;
        ConsumerCallBack callback;

        Consumer(){}
        Consumer(const std::string& ctag, const std::string queue_name, bool ack, const ConsumerCallBack& cb)
            :tag(ctag)
            ,qname(queue_name)
            ,auto_ack(ack)
            ,callback(cb)
        {}
    };

    class QueueConsumer{
    public:
        using ptr = std::shared_ptr<QueueConsumer>;
        QueueConsumer(const std::string qname):_qname(qname), _rr_index(0){}
        Consumer::ptr create(const std::string& ctag, const std::string queue_name, bool ack, const ConsumerCallBack& cb){
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto& c:_consumers){
                if(c->tag == ctag)
                    return Consumer::ptr();
            }
            Consumer::ptr cp = std::make_shared<Consumer>(ctag, queue_name, ack, cb);
            _consumers.push_back(cp);
            return cp;
        }
        void remove(const std::string& ctag){
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto it = _consumers.begin(); it != _consumers.end(); ++it){
                if((*it)->tag == ctag){
                    _consumers.erase(it);
                    return;
                }
            }
        }
        Consumer::ptr choose(){
            std::unique_lock<std::mutex> lock(_mutex);
            if(_consumers.empty()) return Consumer::ptr();
            auto ret = _consumers[_rr_index++];
            _rr_index %= _consumers.size();
            return ret;
        }
        bool exists(const std::string& ctag){
            std::unique_lock<std::mutex> lock(_mutex);
            for(int i = 0; i < _consumers.size(); i++){
                if(_consumers[i]->tag == ctag){
                    return true;
                }
            }
            return false;
        }
        bool empty(){
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.empty();
        }
        void clear(){
            std::unique_lock<std::mutex> lock(_mutex);
            _consumers.clear();
            _rr_index = 0;
        }
    private:
        std::string _qname;
        std::mutex _mutex;
        size_t _rr_index; // RR轮转下标
        std::vector<Consumer::ptr> _consumers;
    };

    class ConsumerManager{
    public:
        using ptr = std::shared_ptr<ConsumerManager>;
        ConsumerManager(){}
        void initComsumerQueue(const std::string & qname){
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _qconsumers.find(qname);
            if(it != _qconsumers.end()){
                return;
            }
            _qconsumers.insert(std::make_pair(qname, std::make_shared<QueueConsumer>(qname)));
        }
        void destroyComsumerQueue(const std::string & qname){
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.erase(qname);
        }
        Consumer::ptr create(const std::string & qname, const std::string& ctag, bool ack, const ConsumerCallBack& cb){
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end()){
                    DLOG("不存在消息队列 %s", qname.c_str());
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->create(ctag, qname, ack, cb);
        }
        void remove(const std::string & qname, const std::string& ctag){
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end()){
                    DLOG("不存在消息队列 %s", qname.c_str());
                    return;
                }
                qcp = it->second;
            }
            qcp->remove(ctag);
        }
        Consumer::ptr choose(const std::string & qname){
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end()){
                    DLOG("不存在消息队列 %s", qname.c_str());
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->choose();
        }
        bool exists(const std::string & qname, const std::string& ctag){
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end()){
                    DLOG("不存在消息队列 %s", qname.c_str());
                    return false;
                }
                qcp = it->second;
            }
            return qcp->exists(ctag);
        }
        bool empty(const std::string & qname){
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end()){
                    DLOG("不存在消息队列 %s", qname.c_str());
                    return false;
                }
                qcp = it->second;
            }
            return qcp->empty();
        }
        void clear(){
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.clear();
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
    };
}